package com.gitee.ixtf.cqrs.verticle

import com.gitee.ixtf.core.J
import com.gitee.ixtf.cqrs.kotlinx.ComponentConsumerMessage
import com.gitee.ixtf.cqrs.kotlinx.EB_ADDRESS_EVENT_SOURCED_ENTITY_CONSUMER
import com.gitee.ixtf.cqrs.kotlinx.ENTITY_ID_KEY
import com.gitee.ixtf.vertx.verticle.BaseCoroutineVerticle
import io.vertx.core.Future
import io.vertx.core.Promise
import io.vertx.core.impl.ContextInternal
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.kotlin.coroutines.receiveChannelHandler
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch

abstract class EventSourcedEntity<S, E> : BaseCoroutineVerticle() {
  private val channel by lazy { vertx.receiveChannelHandler<Pair<Promise<S>, Iterable<E>>>() }
  protected val entityId: String by lazy { config.getString(ENTITY_ID_KEY) }
  private var currentState: S? = null
  protected val publishEventFilter: suspend (event: E) -> Boolean = { false }

  fun invalidate(): Future<Void> =
      if (J.nonBlank(deploymentID)) vertx.undeploy(deploymentID) else Future.succeededFuture()

  protected fun currentState(): S = currentState!!

  protected abstract suspend fun emptyState(): S

  protected abstract suspend fun applyEvent(event: E): S

  protected abstract suspend fun persist(entity: S)

  override suspend fun start() {
    super.start()
    currentState = emptyState()
    launch { channel.consumeEach { (promise, events) -> handle(promise, events) } }
  }

  private suspend fun handle(promise: Promise<S>, events: Iterable<E>) {
    val before = currentState()
    runCatching {
          events.forEach { event -> currentState = applyEvent(event) }
          persist(currentState())
        }
        .onSuccess { promise.complete(currentState()) }
        .onFailure {
          currentState = before
          promise.fail(it)
        }
  }

  override suspend fun stop() {
     channel.cancel(null)
    // channel.close()
    // TODO: 保存当前状态
    // persist(currentState())
    super.stop()
  }

  protected suspend fun persist(events: Collection<E>): S {
    val ctx = vertx.orCreateContext as ContextInternal
    val promise = ctx.promise<S>()
    channel.handle(promise to events)
    return promise.future().coAwait()
  }

  protected suspend fun persist(vararg events: E): S = persist(listOf(*events))

  private suspend fun handleWithPublish(promise: Promise<S>, events: Iterable<E>) {
    val before = currentState()
    runCatching {
          events.forEach { event -> currentState = applyEvent(event) }
          currentState().also { persist(it) }
        }
        .onSuccess { after ->
          promise.complete(after)
          launch {
            val publishEvents = events.filter { publishEventFilter(it) }
            if (J.nonEmpty(publishEvents))
                eventBus.publish(
                    EB_ADDRESS_EVENT_SOURCED_ENTITY_CONSUMER,
                    ComponentConsumerMessage(
                        previous = before,
                        events = publishEvents,
                        current = after,
                    ))
          }
        }
        .onFailure {
          currentState = before
          promise.fail(it)
        }
  }
}
